package cn.doitedu.rtmk.demo4;


import cn.doitedu.rtmk.common.EventBean;
import com.alibaba.fastjson.JSON;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.kafka.clients.consumer.OffsetResetStrategy;

import java.util.HashMap;
import java.util.Map;

/**
 * 相较 demo3的变化： 封装运算机
 * 规则 1： 当 画像标签 age>30 and age<40 AND gender=male[静态画像条件]，在规则上线后，如果该用户发生了 3次+ w行为后 [实时统计画像标签条件] ，再发生 u 行为[触发条件]，立刻推出消息
 * 规则 2： 当 画像标签 active_level=3  AND gender=female[静态画像条件] 在规则上线后，用户依次发生过：(k, b ,c) [实时统计画像标签条件]，当发生 m行为(p1=v1)[触发条件]，推送消息
 */
public class Demo4 {

    public static void main(String[] args) throws Exception {

        // 创建编程入口
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 开启checkpoint
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt");
        env.getCheckpointConfig().setCheckpointTimeout(2000);
        // 设置状态的backend
        env.setStateBackend(new HashMapStateBackend());


        // 构建 kafka source，读取用户实时行为数据
        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers("doitedu:9092")
                .setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.LATEST))
                .setGroupId("doit40-1")
                .setTopics("dwd_events")
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();

        // 读取kafka中的数据为一个流
        DataStreamSource<String> eventsStr = env.fromSource(source, WatermarkStrategy.noWatermarks(), "s");

        // 解析行为日志为javabean
        SingleOutputStreamOperator<EventBean> beanStream = eventsStr.map(json -> JSON.parseObject(json, EventBean.class));


        // 把相同用户的行为，发到相同的subtask去处理
        KeyedStream<EventBean, Long> keyedStream = beanStream.keyBy(bean -> bean.getUid());

        // 核心处理逻辑
        keyedStream.process(new KeyedProcessFunction<Long, EventBean, String>() {

            // 构造一个运算机池
            final HashMap<String, RuleCalculator> calculatorPool = new HashMap<>();

            @Override
            public void open(Configuration parameters) throws Exception {

                // 构造规则1的运算机对象，并初始化
                Rule_1_Calculator rule1Calculator = new Rule_1_Calculator();
                rule1Calculator.init(getRuntimeContext());
                calculatorPool.put("rule-1", rule1Calculator);

                // 构造规则2的运算机对象，并初始化
                Rule_2_Calculator rule2Calculator = new Rule_2_Calculator();
                rule2Calculator.init(getRuntimeContext());
                calculatorPool.put("rule-2", rule2Calculator);

                // 构造规则3的运算机对象，并初始化
                Rule_3_Calculator rule3Calculator = new Rule_3_Calculator();
                rule3Calculator.init(getRuntimeContext());
                calculatorPool.put("rule-3", rule3Calculator);


            }

            @Override
            public void processElement(EventBean eventBean, KeyedProcessFunction<Long, EventBean, String>.Context context, Collector<String> collector) throws Exception {

                // 遍历运算机池中的每一个运算机对象，对当前收到的用户行为数据进行处理
                for (Map.Entry<String, RuleCalculator> entry : calculatorPool.entrySet()) {
                    RuleCalculator calculator = entry.getValue();
                    calculator.calc(eventBean, collector);
                }
            }
        }).print();

        env.execute();
    }
}
